package sunyu.tools.hbase.mapper;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import sunyu.tools.hbase.mapper.pojo.Family;
import sunyu.tools.hbase.mapper.pojo.Row;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;


/**
 * Created by SunYu
 */
public class HbaseRowMapper {

    /**
     * result转换成row对象
     *
     * @param result
     * @param returnRowKeyOnly row对象里面只有rowKey
     *
     * @return
     */
    public static Row mapRow(Result result,
                             Boolean returnRowKeyOnly) {
        List<Row> rows = mapRow(Arrays.asList(result), returnRowKeyOnly);
        if (rows != null && rows.size() > 0) {
            return rows.get(0);
        }
        return null;
    }

    public static List<Row> mapRow(List<Result> results,
                                   Boolean returnRowKeyOnly) {
        if (results != null && results.size() > 0) {
            List<Row> rows = new ArrayList<>();
            for (Result result : results) {
                if (result == null || result.getRow() == null) {
                    return null;
                }
                if (returnRowKeyOnly == null) {
                    returnRowKeyOnly = false;
                }
                Row row = new Row();// 一行
                row.setRowName(Bytes.toString(result.getRow()));
                if (returnRowKeyOnly != null && returnRowKeyOnly == false) {
                    Map<String, Family> familys = row.getFamilys();// 一行里的family集合
                    for (Cell cell : result.rawCells()) {
                        String familyName = Bytes.toString(CellUtil.cloneFamily(cell));// 获取当前列的familyName
                        String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));// 列名
                        String value = Bytes.toString(CellUtil.cloneValue(cell));// 值

                        Family family = familys.get(familyName);
                        if (family == null) {
                            family = new Family();
                            familys.put(familyName, family);
                        }

                        Map<String, Object> columns = family.getColumns();// 当前family的所有列
                        columns.put(columnName, value);
                    }
                }
                rows.add(row);
            }
            return rows;
        }
        return null;
    }

}
